package org.qee.mq.custom;

import lombok.Data;

import java.io.File;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.locks.ReentrantLock;
import java.util.stream.Collectors;

import static org.qee.mq.custom.MappedFile.MAPPED_FILE_SIZE;

/**
 * @ProjectName: learning
 * @Package: org.qee.mq.custom
 * @ClassName: DefaultQueue
 * @Description:
 * @Date: 2021/12/20 10:43 上午
 * @Version: 1.0
 */
public class DefaultQueue implements Queue {

    private String topicName;

    //1 M  大小
    public static int MSG_SIZE = 1024 * 1024;


    // 一个 MappedFile 一个G
    private List<MappedFile> mappedFileList = new CopyOnWriteArrayList<>();

    private ReentrantLock lock = new ReentrantLock();

    private String parentPath;


    private static AtomicLong atomicLong = new AtomicLong(0);

    public DefaultQueue(String topicName, String parentPath) {
        this.topicName = topicName;
        this.parentPath = parentPath;
    }

    @Override
    public boolean offer(Message message) {
        if (message.getSize() > MSG_SIZE) {
            return false;
        }
        MappedFile mappedFile = null;
        try {
            mappedFile = getLastMappedFile(true);
        } catch (IOException e) {
            e.printStackTrace();
            return false;
        }
        try {
            lock.lock();
            MappedFile.WriteResult result = mappedFile.appendMsg(message);
            if (result == MappedFile.WriteResult.OK) {
                return true;
            }
            if (result == MappedFile.WriteResult.END_OF_FILE) {
                MappedFile newMappedFile = createLastMappedFile(topicName + "-" + atomicLong.getAndAdd(MAPPED_FILE_SIZE));
                MappedFile.WriteResult appendMsg = newMappedFile.appendMsg(message);
                return appendMsg == MappedFile.WriteResult.OK;
            }
        } catch (IOException e) {
            e.printStackTrace();
        } finally {
            lock.unlock();
        }
        return false;
    }

    @Override
    public TakeResult take(long offset, int msgCount) throws IOException {
        MappedFile.ReadResult readResult = null;
        List<byte[]> messages = new ArrayList<>();
        long beginOffset = offset;
        int allSize = msgCount;
        TakeResult takeResult = new TakeResult();
        do {
            MappedFileResult mappedFileResult = getMappedFileFromBeginOffset(beginOffset);
            if (mappedFileResult.getBeginOffset() == -1) {
                break;
            }
            readResult = mappedFileResult.getMappedFile().read(mappedFileResult.getBeginOffset(), msgCount);
            allSize -= readResult.getSize();
            messages.addAll(readResult.getBytes());
            beginOffset = readResult.getEndOffset() + 1;
        } while (!(readResult.getIndexFile() == mappedFileList.size() - 1) && allSize > 0);
        takeResult.setResult(messages);
        takeResult.setLastOffset(readResult != null ? readResult.getEndOffset() : 0);
        return takeResult;
    }

    @Override
    public void reload() {
        File file = new File(parentPath);
        String[] list = file.list();
        List<String> collect = Arrays.stream(list).sorted().collect(Collectors.toList());
        for (String fileName : collect) {
            if (fileName.startsWith(topicName)) {
                try {
                    MappedFile mappedFile = new MappedFile(fileName);
                    mappedFileList.add(mappedFile);
                    mappedFile.recreateIndex();
                } catch (IOException ioException) {
                    ioException.printStackTrace();
                }

            }
        }


    }

    private MappedFile createLastMappedFile(String name) throws IOException {
        MappedFile mappedFile = new MappedFile(name);
        mappedFileList.add(mappedFile);
        return mappedFile;
    }

    @Data
    static class MappedFileResult {
        private MappedFile mappedFile;
        private long beginOffset;
    }

    private MappedFileResult getMappedFileFromBeginOffset(long offset) {
        int lastIndex = mappedFileList.size() - 1;
        long sy = offset % MAPPED_FILE_SIZE;
        long n = offset / MAPPED_FILE_SIZE;
        MappedFileResult mappedFileResult = new MappedFileResult();
        if (n > lastIndex) {
            mappedFileResult.setBeginOffset(-1);
            return mappedFileResult;
        }

        //  sy+ n*MAPPED_FILE_SIZE==offset
        MappedFile mappedFile = mappedFileList.get((int) n);
        mappedFileResult.setMappedFile(mappedFile);
        mappedFileResult.setBeginOffset(sy);
        return mappedFileResult;
    }

    private MappedFile getLastMappedFile(boolean needCreate) throws IOException {
        MappedFile mappedFile = null;
        if (!mappedFileList.isEmpty()) {
            mappedFile = mappedFileList.get(mappedFileList.size() - 1);
            if (mappedFile != null) {
                return mappedFile;
            }
        }
        if (!needCreate) {
            return null;
        }
        mappedFile = new MappedFile(topicName + "-" + atomicLong.getAndAdd(MAPPED_FILE_SIZE));
        mappedFileList.add(mappedFile);
        return mappedFile;
    }

}
